package com.fxkj.cloud.drms.infra.repository;

import com.fxkj.cloud.dbs.common.constants.SceneLinkageOperationType;
import com.fxkj.cloud.dbs.common.event.stream.RemoteSceneLinkageChangeEvent;
import com.fxkj.cloud.dbs.common.model.kv.TsKvEntry;
import com.fxkj.cloud.dbs.common.model.linkage.FxSceneLinkage;
import com.fxkj.cloud.dbs.common.model.linkage.SceneLinkageEnableStatus;
import com.fxkj.cloud.drms.domain.device.repository.DeviceRepository;
import com.fxkj.cloud.drms.domain.linkage.FxLinkageProcessor;
import com.fxkj.cloud.drms.domain.linkage.FxProcessor;
import com.fxkj.cloud.drms.domain.linkage.LinkageProcessorService;
import com.fxkj.cloud.drms.domain.mqtt.service.SenderService;
import com.fxkj.cloud.drms.domain.ts.service.TimeseriesService;
import com.fxkj.cloud.drms.infra.feignApi.SceneLinkageFeignApi;
import com.fxkj.cloud.drms.infra.persistence.cache.SceneLinkageCacheRepository;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * <p>description: </p>
 * @author ww
 * @date 2025/3/31 14:26
 */
@Component
@Slf4j
public class LinkageProcessorServiceImpl implements LinkageProcessorService, ApplicationListener<ApplicationReadyEvent> {
    private final SceneLinkageCacheRepository sceneLinkageCacheRepository;
    private final SceneLinkageFeignApi sceneLinkageFeignApi;
    private final TimeseriesService timeseriesService;
    private final DeviceRepository deviceRepository;
    //场景联动处理器映射
    private final ConcurrentHashMap<UUID, FxProcessor> linkageProcessorMap = new ConcurrentHashMap<>();
    //包含定时触发器的场景联动处理器映射
    private final ConcurrentHashMap<UUID, FxProcessor> scheduleLinkageProcessorMap = new ConcurrentHashMap<>();
    private final SenderService sender;
    private final ApplicationEventPublisher eventPublisher;
    private final StreamBridge streamBridge;

    public LinkageProcessorServiceImpl(SceneLinkageCacheRepository sceneLinkageCacheRepository, SceneLinkageFeignApi sceneLinkageFeignApi, TimeseriesService timeseriesService, DeviceRepository deviceCacheRepository, SenderService sender, ApplicationEventPublisher eventPublisher, StreamBridge streamBridge) {
        this.sceneLinkageCacheRepository = sceneLinkageCacheRepository;
        this.sceneLinkageFeignApi = sceneLinkageFeignApi;
        this.timeseriesService = timeseriesService;
        this.deviceRepository = deviceCacheRepository;
        this.sender = sender;
        this.eventPublisher = eventPublisher;
        this.streamBridge = streamBridge;
    }

    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        //初始化场景联动处理器
        List<FxSceneLinkage> fxSceneLinkages = sceneLinkageFeignApi.listInfo(SceneLinkageEnableStatus.STARTED.getCode());
        //缓存场景id->场景处理器关系
        Map<UUID, FxLinkageProcessor> collect = fxSceneLinkages.stream()
                .map(fxSceneLinkage ->
                        new FxLinkageProcessor(fxSceneLinkage, timeseriesService, sender, deviceRepository, eventPublisher, streamBridge))
                .collect(Collectors.toMap(FxLinkageProcessor::getSceneId, processor -> processor));
        linkageProcessorMap.putAll(collect);

        //缓存定时触发器场景id->场景处理器关系
        fxSceneLinkages.stream().filter(FxSceneLinkage::isSchedule).forEach(fxSceneLinkage -> {
            scheduleLinkageProcessorMap.put(fxSceneLinkage.getId(), collect.get(fxSceneLinkage.getId()));
        });
    }

    @Override
    public void process(UUID deviceId, List<TsKvEntry> deviceKvs) {
        //获取和设备关联的场景联动id
        Set<UUID> sceneLinkageIds = sceneLinkageCacheRepository.getSceneLinkageIdsByDeviceId(deviceId);
        if (null == sceneLinkageIds || sceneLinkageIds.isEmpty()) {
            return;
        }
        sceneLinkageIds.forEach(sceneLinkageId -> {
            linkageProcessorMap.get(sceneLinkageId).processor(deviceId, deviceKvs);
        });
    }

    @Override
    public void sceneLinkageExecute(UUID sceneLinkageId, List<TsKvEntry> deviceKvs) {
        FxProcessor fxProcessor = linkageProcessorMap.get(sceneLinkageId);
        if (null == fxProcessor) {
            return;
        }
        fxProcessor.execute(deviceKvs);
    }

    /**
     * 场景联动变更事件处理
     * 1.删除事件
     * 清理当前的 场景联动处理器映射
     * 2.新增/更新事件
     * 首先是 覆盖旧的处理器
     * 然后是 添加新的设备-> 场景联动缓存 移除旧的设备-> 场景联动缓存
     * 最后是 销毁旧的处理器,停止任务,然后创建新的处理器,启动任务
     *
     * @param event
     */
    @EventListener
    public void onSceneLinkageChange(RemoteSceneLinkageChangeEvent event) {
        //获取变动类型
        SceneLinkageOperationType operationType = event.getOperationType();
        UUID sceneLinkageId = event.getSceneLinkageId();
        FxSceneLinkage fxSceneLinkage = event.getFxSceneLinkage();
        if (null == operationType) {
            return;
        }
        //场景联动删除/更新/停止事件
        //停止处理器,移除之前的缓存信息
        if (SceneLinkageOperationType.DELETE.equals(operationType)
                || SceneLinkageOperationType.UPDATE.equals(operationType)
                || SceneLinkageOperationType.STOP.equals(operationType)) {
            log.trace("[LinkageProcessorServiceImpl] -| 场景联动处理器变更类型为{}", operationType.name());
            //销毁处理器并移除缓存
            FxProcessor fxProcessor = linkageProcessorMap.get(sceneLinkageId);
            if (null == fxProcessor) {
                log.warn("[LinkageProcessorServiceImpl] -| 删除/更新/停止场景联动信息不存在");
                return;
            }
            //成功与否都先移除缓存
            removeProcessor(sceneLinkageId);
            //移除设备场景联动缓存
            sceneLinkageCacheRepository.removeSceneLinkageBySceneLinkageId(sceneLinkageId);
            //销毁任务
            destroyProcessor(fxProcessor);
        }
        //场景联动启动事件
        else if (SceneLinkageOperationType.START.equals(operationType)) {
            //获取场景联动信息
            //如果场景联动不存在,则创建
            if (null == fxSceneLinkage) {
                log.warn("[LinkageProcessorServiceImpl] -| 创建/更新/启动场景联动信息不存在");
                return;
            }
            //若存在处理器,则是更新操作,先销毁旧处理器,停止任务,然后创建新处理器,启动任务
            //覆盖旧的处理器
            putProcessor(sceneLinkageId, fxSceneLinkage);
            //移除设备场景联动缓存
            sceneLinkageCacheRepository.removeSceneLinkageBySceneLinkageId(sceneLinkageId);
            //添加新的设备-> 场景联动缓存
            sceneLinkageCacheRepository.addSceneLinkageByDeviceId(fxSceneLinkage.getDeviceIds(), fxSceneLinkage);
        }
        //其他情况 暂无
        else {
            log.warn("[LinkageProcessorServiceImpl] -| 场景联动状态变更时间类型暂不支持,不做处理");
        }
    }

    private void removeProcessor(UUID sceneLinkageId) {
        this.linkageProcessorMap.remove(sceneLinkageId);
        this.scheduleLinkageProcessorMap.remove(sceneLinkageId);
    }

    private void putProcessor(UUID sceneLinkageId, FxSceneLinkage linkage) {
        FxLinkageProcessor processor = new FxLinkageProcessor(linkage, timeseriesService, sender, deviceRepository, eventPublisher, streamBridge);
        this.linkageProcessorMap.put(sceneLinkageId, processor);
        if (linkage.isSchedule()) {
            this.scheduleLinkageProcessorMap.put(sceneLinkageId, processor);
        }
    }

    /**
     * 销毁处理器
     * @param processor
     */
    public void destroyProcessor(FxProcessor processor) {
        if (null == processor) {
            return;
        }
        ListenableFuture<?> destroyFuture = Futures.whenAllComplete(processor.destroy()).run(() -> {
            log.trace("[LinkageProcessorServiceImpl] -| 场景联动处理器销毁成功,移除缓存");
        }, MoreExecutors.directExecutor());
        Futures.addCallback(destroyFuture, new FutureCallback<Object>() {
            @Override
            public void onSuccess(Object result) {
            }

            @Override
            public void onFailure(Throwable t) {
                log.error("[LinkageProcessorServiceImpl] -| 场景联动处理器销毁失败/等待超时", t);
            }
        }, MoreExecutors.directExecutor());
    }

}
